package com.message.queue.rocket.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

/**
 * Created by 张波 on 2019/10/14.
 */
@Slf4j
public class ConsumerParent {

    public static final String CONSUMER_GROUP = "default_push_consumer";

    public static final String TOPIC = "push_topic";

    public static MessageListenerConcurrently getListener() {
        return new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt messageExt = msgs.get(0);
                try {
                    String body = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                    System.out.println("topic:" + messageExt.getTopic() + ",tags:" + messageExt.getTags() + ",keys:" + messageExt.getKeys()
                            + ",body:" + body);
                } catch (Exception e) {
                    log.error("异常：{}", e);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        };
    }
}
